package cn.gemiman.daoyi.module.iot.open.jhl.kafka.listener;

import cn.gemiman.daoyi.module.iot.controller.admin.kafkamsglogs.vo.KafkaMsgLogsSaveReqVO;
import cn.gemiman.daoyi.module.iot.service.kafkamsglogs.KafkaMsgLogsService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class AihujiankangKafkaConsumerListener {
    @Resource
    private KafkaMsgLogsService equipmentMsgLogsService;
    @Value("${spring.kafka.aihujiankang.consumer.group-id}")
    private String groupId;
    @Value("${spring.kafka.aihujiankang.bootstrap-servers}")
    private String bootstrapServers;

    @KafkaListener(topics = "jhl-dw-topic-ED767-DY-001", containerFactory = "aihujiankangKafkaContainerFactory")
    public void listenUser(ConsumerRecord<?, String> record, Acknowledgment acknowledgment) {
        try {
            String body = record.value();
            equipmentMsgLogsService.createKafkaMsgLogs(
                    new KafkaMsgLogsSaveReqVO()
                            .setMsg(body)
                            .setTopics("jhl-dw-topic-ED767-DY-001")
                            .setBootstrapServers(bootstrapServers)
                            .setConsumerGroupId(groupId)
                            .setViewed(false)
            );
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            acknowledgment.acknowledge();
        }
    }
}
